package com.shujia.sink;

import com.shujia.common.Config;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.http.HttpHost;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestClientBuilder;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.xcontent.XContentType;

import java.util.HashMap;
import java.util.Map;

public class ESSink extends RichSinkFunction<String> {

    private RestHighLevelClient client;

    @Override
    public void open(Configuration parameters) throws Exception {

        // 创建es连接
        HttpHost httpHost = HttpHost.create(Config.getString("es.connect"));
        RestClientBuilder builder = RestClient.builder(httpHost);
        client = new RestHighLevelClient(builder);

    }

    @Override
    public void invoke(String value, Context context) throws Exception {
        Map<String, String> json = new HashMap<>();

        String[] split = value.split("\\|");

        json.put("id", split[0]);
        json.put("sent_id", split[1]);
        json.put("user_id", split[2]);
        json.put("source", split[3]);
        json.put("reposts_count", split[4]);
        json.put("comments_count", split[5]);
        json.put("attitudes_count", split[6]);
        json.put("text", split[7]);

        IndexRequest request = new IndexRequest("wz").id(split[0]);
        request.source(json, XContentType.JSON);

        client.index(request, RequestOptions.DEFAULT);

    }
}
